package com.wang.event;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;

public class SerialEventManager<T> {
    private EventManager eventManager;
    private BlockingQueue<Event<T>> queue = new LinkedBlockingQueue<>();
    private Map<Consumer<Event<T>>, Boolean> statusMap = new HashMap<>();

    public SerialEventManager(EventManager eventManager) {
        this.eventManager = eventManager;
    }

    public void register(Consumer<Event<T>> consumer) {
        this.statusMap.put(consumer, false);
        eventManager.register(new CallbackConsumer(consumer));
    }

    public void sendEvent(Event<T> event) {
        this.queue.add(event);
    }

    private void callback(Consumer<Event<T>> consumer) {
        statusMap.put(consumer, true);
        boolean allOk = statusMap.values().stream().allMatch(flag -> flag);
        if (!allOk) {
            return;
        }
        // 全部线程消费完
        statusMap.entrySet().forEach(entry -> {
            entry.setValue(false);
        });
        new Thread(() -> {
            try {
                eventManager.sendEvent(this.queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }

    private class CallbackConsumer implements Consumer<Event<T>> {
        private Consumer<Event<T>> consumer;

        public CallbackConsumer(Consumer<Event<T>> consumer) {
            this.consumer = consumer;
        }

        @Override
        public void accept(Event<T> event) {
            this.consumer.accept(event);
            SerialEventManager.this.callback(this);
        }
    }
}
